use async_std::net::TcpStream;
use futures::{SinkExt, StreamExt};
use honeybee::prelude::*;
use honeybee::Result;
use honeybee_proto::hall;
use prost::Message as ProstMessage;

use crate::message_register::message_process;
use crate::user::{Disconnected, User};

pub struct Connect {
    addr: Option<Addr<Connect>>,
    account: String,
    server_addr: String,
    user_addr: Option<Addr<User>>,
    framed_write: Option<FramedWrite<Compat<TcpStream>, UserMsgCodec>>,
}

impl Connect {
    pub fn new(account: &str, server_addr: &str) -> Self {
        Self {
            addr: None,
            account: account.into(),
            server_addr: server_addr.into(),
            user_addr: None,
            framed_write: None,
        }
    }

    async fn send<T: 'static + prost::Message + MessageId>(&mut self, msg: &T) -> Result<()> {
        if let Some(ref mut write) = self.framed_write {
            return write.send(msg).await;
        }
        Ok(())
    }
}

#[async_trait::async_trait]
impl Actor for Connect {
    fn name(&self) -> &str {
        "Connect"
    }

    async fn started(&mut self, ctx: &mut Context<Self>) -> honeybee::Result<()> {
        let addr = ctx.address();
        let _ = addr.send(Login);
        info!("robot:{} started", addr.actor_id());
        self.addr = Some(addr);
        Ok(())
    }

    async fn stopped(&mut self, ctx: &mut Context<Self>) {
        if let Some(addr) = self.user_addr.take() {
            let _ = addr.send(Disconnected);
        }
        info!(
            "{} actor:{} stoped, disconnect",
            self.name(),
            ctx.actor_id(),
        );
    }
}

#[message]
struct Login;

#[async_trait::async_trait]
impl Handler<Login> for Connect {
    async fn handle(&mut self, ctx: &mut Context<Self>, _: Login) -> Result<()> {
        let addr = self.addr.take().ok_or_else(|| anyhow!("inner error"))?;

        let stream = TcpStream::connect(&self.server_addr).await?;
        let peer_addr = stream
            .peer_addr()
            .map(|addr| addr.to_string())
            .unwrap_or_else(|_| "None".into());

        if let Err(err) = stream.set_nodelay(true) {
            error!("user stream {:?} set_nodelay err:{}", stream, err);
        }

        self.framed_write = Some(FramedWrite::new(
            stream.clone().compat(),
            UserMsgCodec::new(),
        ));
        let mut framed_read = FramedRead::new(stream.clone().compat(), UserMsgCodec::new());

        self.send(&hall::ReqLogin {
            token: self.account.clone(),
        })
        .await?;

        let rsp_login_msg = framed_read
            .next()
            .await
            .ok_or_else(|| anyhow!("login failed"))??;
        let msg = hall::RspLogin::decode(rsp_login_msg.bytes)
            .map_err(|err| anyhow!("RspLogin failed, err:{}", err))?;

        self.user_addr = Some(
            User::new(self.account.clone(), msg.uid, addr, peer_addr)
                .start()
                .await?,
        );

        ctx.add_stream(framed_read);
        Ok(())
    }
}

type MsgResult = std::result::Result<UserRawMsg, anyhow::Error>;
#[async_trait::async_trait]
impl StreamHandler<MsgResult> for Connect {
    async fn handle(&mut self, ctx: &mut Context<Self>, msg: MsgResult) {
        match msg {
            Ok(msg) => {
                let msg_id = msg.id;
                if let Some(ref user_addr) = self.user_addr {
                    if let Err(err) = message_process(user_addr, msg) {
                        error!("{} process msg:{} error:{}", self.name(), msg_id, err)
                    }
                } else {
                    error!(
                        "{} process msg:{} error, user_addr is None",
                        self.name(),
                        msg_id
                    )
                }
            }
            Err(err) => {
                error!("{} receive message error:{}", self.name(), err);
                ctx.stop(None);
            }
        }
    }

    async fn finished(&mut self, ctx: &mut Context<Self>) {
        error!("{} disconnected", self.name());
        let _ = ctx.stop(None);
    }
}

pub struct SendUserMsg<T: 'static + MessageId + prost::Message>(pub T);

impl<T: 'static + MessageId + prost::Message> Message for SendUserMsg<T> {
    type Result = ();
}

#[async_trait::async_trait]
impl<T: 'static + MessageId + prost::Message> Handler<SendUserMsg<T>> for Connect {
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SendUserMsg<T>) -> Result<()> {
        if let Err(err) = self.send(&msg.0).await {
            error!("{} send msg err:{}", self.name(), err);
            return Ok(());
        }
        Ok(())
    }
}
